package cn.wanda.utils.modelGroup

import cn.wanda.utils.config.groupJobTrait
import org.apache.spark.sql.{Row, SparkSession}

class createOdIntvlMax extends groupJobTrait {

    /**
      * od_intvl_max创建
      */
    def create(sparkSessionContext: SparkSession): Unit = {
        val odIntvlMaxMap = sparkSessionContext.sql(odIntvlMaxSql)
          .rdd
          .map(x => {
              val member_code = x.get(0).toString
              val payment_list = x.getSeq(1).asInstanceOf[Seq[String]].toArray

              var max = -99

              for (i <- payment_list.indices) {
                  if (i < payment_list.length - 1) {
                      val payment = payment_list(i + 1)
                      val payment_new = payment_list(i)
                      val cha = diffDay(payment, payment_new)
                      if (cha > max) {
                          max = cha.toInt
                      }
                  }
              }

              (member_code, max)
          })

        val odIntvlMaxRdd = odIntvlMaxMap.map(
            x => Row(x._1, x._2)
        )

        sparkSessionContext.sqlContext
          .createDataFrame(odIntvlMaxRdd, odIntvlMaxSchema)
          .createTempView("od_intvl_max")

    }
}
